package rabbitmq

import (
	"errors"
	"fmt"
	"net/url"

	"git.100tal.com/wangxiao_dahai_backend/arch-nova/middleware"
	"github.com/assembla/cony"
	"github.com/streadway/amqp"
)

// 单个消费者或生产者
type RabbitMq struct {
	cli      *cony.Client
	Message  chan string        // 发送的消息
	Delivery chan amqp.Delivery // 消费的消息由业务控制，是否确认、拒绝等
	Error    chan error         // 错误必须处理，如打印到日志等
}

func NewRabbitMq(user, password, host string, port int) *RabbitMq {
	addr := fmt.Sprintf("amqp://%s:%s@%s:%d", url.QueryEscape(user), url.QueryEscape(password), host, port)
	return &RabbitMq{
		cli: cony.NewClient(
			cony.URL(addr),
			cony.Backoff(cony.DefaultBackoff),
		),
		Message:  make(chan string),
		Delivery: make(chan amqp.Delivery),
		Error:    make(chan error, 10),
	}
}

// 多个消费者或生产者包装
type MqWrapper struct {
	cli   *cony.Client
	Error chan error // 错误必须处理，如打印到日志等
}

func NewMqWrapper(user, password, host string, port int) *MqWrapper {
	addr := fmt.Sprintf("amqp://%s:%s@%s:%d", url.QueryEscape(user), url.QueryEscape(password), host, port)
	return &MqWrapper{
		cli: cony.NewClient(
			cony.URL(addr),
			cony.Backoff(cony.DefaultBackoff),
		),
	}
}

func (mq *MqWrapper) Loop() bool {
	return mq.cli.Loop()
}

// 开始处理，异常错误等
func (mq *MqWrapper) Dispatch() {
	// Client loop sends out declarations(exchanges, queues, bindings
	// etc) to the AMQP server. It also handles reconnecting.
	//
	go func() {
		defer func() {
			if err := recover(); err != nil {
				middleware.LogError(nil, "[RecoveryPanic] [RabbitMq] "+err.(string))
			}
		}()

		for mq.cli.Loop() {
			select {
			case err := <-mq.cli.Errors():
				e := errors.New("amqp connection error: " + err.Error())
				mq.Error <- e

			case blocked := <-mq.cli.Blocking():
				e := errors.New("amqp connection blocked: " + blocked.Reason)
				mq.Error <- e
			}
		}
	}()
}
